package com.juma.cartridge.rlock;

import java.util.UUID;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.util.Pool;

import com.juma.cartridge.rlock.exception.LockException;
import com.juma.cartridge.rlock.fair.FairLock;
import com.juma.cartridge.rlock.fair.LockQueue;

/**
 * 基于redis实现的分布式锁
 * @ClassName RedisLockManager
 * @Description
 * @Author zhuangguangyou
 * @Date 2019年7月24日 下午3:53:30
 * @Version 1.0
 *
 */
public class RedisLockManager implements LockManager {
	private static final String PRE = "rlock";
	private static ExecutorService executor;
	
	private redis.clients.jedis.util.Pool<Jedis> jedisPool;
	private DelayQueue<RLock> leaseQueue;
	private LeaseTask leaseTask;
	
	
	public RedisLockManager(Pool<Jedis> pool) {
		this.jedisPool = pool;
		executor = Executors.newSingleThreadExecutor();
		leaseQueue = new DelayQueue<RLock>();
	}
	
	public void destory() {
		leaseTask.close();
		leaseQueue.clear();
	}
	
	@Override
	public Lock getLock(String key,int lockSeconds,boolean fair) {
		if (key == null || key.isEmpty()) {
			throw new IllegalArgumentException(
					"argument 'key' can not be empty");
		}
		if (lockSeconds < 1) {
			throw new IllegalArgumentException(
					"argument 'lockSeconds' must greater than 0");
		}
		//续租线程初始化
		if(leaseTask==null) {
			synchronized (this) {
				if(leaseTask==null) {
					leaseTask = new LeaseTask(leaseQueue);
					executor.execute(leaseTask);		
				}
			}
		}
		if(!leaseTask.isRunning()) {
			throw new LockException("this lock manager have already been destroyed!");
		}
		return buildLock(key,lockSeconds,fair);
	}

	private Lock buildLock(String key,int lockSeconds,boolean fair) {
		Lock lock = LockHolder.getLock(key);
		if(lock==null) {
			lock = createLock(key,lockSeconds,fair);
			LockHolder.putLock(key, lock);
		}
		return lock;
	}
	
	private Lock createLock(String key, int lockSeconds,boolean fair) {
		String id = UUID.randomUUID().toString().replaceAll("-", "");
		long expireTime = System.currentTimeMillis()+lockSeconds*1000;
		LockNode node = new LockNode(key,id,expireTime,lockSeconds);
		Lock lock = null;
		if(fair) {
			LockQueue lq = new LockQueue(PRE+".fair."+key,jedisPool);
			lq.appendNode(node);
			lock = new FairLock(this,node,Thread.currentThread(),lq);	
		} else {
			lock = new NonFairLock(this,node,Thread.currentThread(),PRE+".nfair."+key,jedisPool);
		}
		return lock;
	}
	
	public void lockAcquired(String key) {
		Lock lock = LockHolder.getLock(key);
		if (lock!=null) {
			leaseQueue.add((RLock)lock);
		}
	}
	
	public void lockOver(String key) {
		LockHolder.removeLock(key);
	}
	
}
